package org.qiyu.live.user.provider.service.impl;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Maps;
import jakarta.annotation.Resource;
import org.apache.rocketmq.client.producer.MQProducer;
import org.apache.rocketmq.common.message.Message;
import org.idea.qiyu.live.framework.redis.starter.key.UserProviderCacheKeyBuilder;
import org.qiyu.live.common.interfaces.utils.ConvertBeanUtils;
import org.qiyu.live.user.dto.UserDTO;
import org.qiyu.live.user.provider.dao.mapper.IUserMapper;
import org.qiyu.live.user.provider.dao.po.UserPO;
import org.qiyu.live.user.provider.service.IUserService;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.SessionCallback;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import static org.qiyu.live.user.constants.UserProviderTopicNames.CHANGE_TOPIC;

@Service
public class UserServiceImpl implements IUserService {

    @Resource()
    private IUserMapper iUserMapper;

    @Resource
    private RedisTemplate<String,UserDTO> redisTemplate;

    @Resource
    private UserProviderCacheKeyBuilder userProviderCacheKeyBuilder;

    @Resource
    MQProducer mqProducer;

    @Override
    public UserDTO getByUserId(Long userId) {
        if (userId == null) {
            return null;
        }
        String key = userProviderCacheKeyBuilder.buildUserInfoKey(userId);
        UserDTO userDTO = redisTemplate.opsForValue().get(key);
        if (userDTO != null) {
            return userDTO;
        }
        UserDTO convert = ConvertBeanUtils.convert(iUserMapper.selectById(userId), UserDTO.class);
        if(convert != null){
            redisTemplate.opsForValue().set(key,convert,30, TimeUnit.SECONDS);
        }
        return convert;
    }

    @Override
    public boolean updateUserInfo(UserDTO userDTO) {
        if(userDTO == null){return false;}
        iUserMapper.updateById(ConvertBeanUtils.convert(userDTO,UserPO.class));
        String key = userProviderCacheKeyBuilder.buildUserInfoKey(userDTO.getUserId());
        //立即删除缓存
        redisTemplate.delete(key);
        try {
            //发送延迟消息，触发二次删除缓存操作
            Message message = new Message();
            message.setBody(JSON.toJSONString(userDTO).getBytes());
            message.setTopic(CHANGE_TOPIC);
            //延迟级别，1代表延迟一秒发送
            message.setDelayTimeLevel(1);
            mqProducer.send(message);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return true;
    }

    @Override
    public boolean insertOne(UserDTO userDTO) {
        if(userDTO == null){return false;}
        iUserMapper.insert( ConvertBeanUtils.convert(userDTO, UserPO.class));
        return true;
    }

    @Override
    public Map<Long, UserDTO> batchQueryUserInfo(List<Long> userIdList) {
        if(CollectionUtils.isEmpty(userIdList)){
            return Maps.newHashMap();
        }
        userIdList = userIdList.stream().filter(id -> id > 10000).collect(Collectors.toList());
        if(CollectionUtils.isEmpty(userIdList)){
            return Maps.newHashMap();
        }
        //处理数据前缀
        List<String> keyList = new ArrayList<>();
        for(Long userId : userIdList){
            keyList.add(userProviderCacheKeyBuilder.buildUserInfoKey(userId));
        }
        //查询缓存，把能从缓存查到的的数据装起来，查不到的数据保存去数据库查询
        List<UserDTO> userDTOS = redisTemplate.opsForValue().multiGet(keyList).stream().filter(x -> x != null).collect(Collectors.toList());
        //如果所有数据都在缓存中能查询到就全部返回
        if(!CollectionUtils.isEmpty(userDTOS) && userIdList.size() == userDTOS.size() ) {
            return userDTOS.stream().collect(Collectors.toMap(UserDTO::getUserId, x -> x));
        }
        //获取缓存中不存在的用户id
        List<Long> cacheList = userDTOS.stream().map(UserDTO::getUserId).toList();
        List<Long> notInList = userIdList.stream().filter(x -> !cacheList.contains(x)).toList();

        //使用多线程分组，批量查询
        Map<Long, List<Long>> userIdMap = notInList.stream().collect(Collectors.groupingBy(userId -> userId % 100));
        List<UserDTO> dbSelectList = new CopyOnWriteArrayList<>();
        userIdMap.values().parallelStream().forEach(idList -> {
            List<UserPO> userPOS = iUserMapper.selectBatchIds(idList);
            dbSelectList.addAll(ConvertBeanUtils.convertList(userPOS, UserDTO.class));
        });
        //判断查询结果是不是存在，查到数据将数据放入缓存
        if(!CollectionUtils.isEmpty(dbSelectList)){
            Map<String,UserDTO> saveCacheMap = dbSelectList.stream().collect(Collectors.toMap(userDTO -> userProviderCacheKeyBuilder.buildUserInfoKey(userDTO.getUserId()), x -> x));
            redisTemplate.opsForValue().multiSet(saveCacheMap);
            //使用管道批量传输命令，减少io开销
            redisTemplate.executePipelined(new SessionCallback<Object>() {
                @Override
                public <K, V> Object execute(RedisOperations<K, V> operations) throws DataAccessException {
                    for(String key : saveCacheMap.keySet()) {
                        operations.expire((K)key, createRandomExpireTime(),TimeUnit.SECONDS);
                    }
                    return null;
                }
            });
            userDTOS.addAll(dbSelectList);
        }
        return userDTOS.stream().distinct().collect(Collectors.toMap(UserDTO::getUserId, x -> x));
    }
    public int createRandomExpireTime(){
        int time = ThreadLocalRandom.current().nextInt(10000);
        return time + 60 * 30;
    }

}
